package com.base.listen;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import com.receive.gate.message.ReceivedGateMessage;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.DelimiterBasedFrameDecoder;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
//import io.netty.handler.logging.LoggingHandler;

public class ServerListen {
	private final int port;
	//安全消息队列 缓存消息
	private static  ConcurrentLinkedQueue<BaseServiceSession> reciveMessageQueue ;
	private static ConcurrentLinkedQueue<BaseServiceSession> sendMessageQueue;
    
    public ServerListen(int port) {
        this.port = port;
        reciveMessageQueue = new ConcurrentLinkedQueue<BaseServiceSession>();
        sendMessageQueue = new ConcurrentLinkedQueue<BaseServiceSession>();
    }
    
    
    public void run() {
        EventLoopGroup bossGroup = new NioEventLoopGroup();        // 用来接收进来的连接
        EventLoopGroup workerGroup = new NioEventLoopGroup();    // 用来处理已经被接收的连接
        System.out.println("准备运行端口：" + port);
        
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)            // 这里告诉Channel如何接收新的连接
            .childHandler( new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                	
                	ChannelPipeline pipeline = ch.pipeline();
                    ByteBuf byteBuf = Unpooled.copiedBuffer("$_".getBytes());
                    pipeline.addLast(new DelimiterBasedFrameDecoder(1024,byteBuf));
                    pipeline.addLast(new StringDecoder());
                    pipeline.addLast(new StringEncoder());
	                //pipeline.addLast("logging",new LoggingHandler("DEBUG"));//设置log监听器，并且日志级别为debug，方便观察运行流程
                	// 自定义处理类
                    pipeline.addLast(new ServerListenHandler());
                }
            })
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true);
            
            // 绑定端口，开始接收进来的连接
            ChannelFuture f = b.bind(port).sync();
            ExecutorService singleThreadExecutorServiceReceive = Executors.newSingleThreadExecutor();
            singleThreadExecutorServiceReceive.execute(() -> {
				while(true){
					if(!reciveMessageQueue.isEmpty()) {
						BaseServiceSession baseServiceSession = reciveMessageQueue.poll();
						ReceivedGateMessage.Read(baseServiceSession);
					
					}else {
						 try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
			});
			
            ExecutorService singleThreadExecutorServiceSend = Executors.newSingleThreadExecutor();
            singleThreadExecutorServiceSend.execute(() -> {
				while(true){
					if(!sendMessageQueue.isEmpty()) {
						BaseServiceSession baseServiceSession = sendMessageQueue.poll();
						baseServiceSession.getCtx().write(baseServiceSession.getMsgString()+"$_");				
						baseServiceSession.getCtx().flush();
					
					}else {
						 try {
							Thread.sleep(500);
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
				}
			});
            // 等待服务器socket关闭
            f.channel().closeFuture().sync();
        } catch (Exception e) {
            workerGroup.shutdownGracefully();
            bossGroup.shutdownGracefully();
        }
    }
    
	public static void reciveMessage(ChannelHandlerContext ctx, String msg) {
		//只是负责向消息队列中丢消息
		BaseServiceSession serviceSession = new BaseServiceSession(ctx, msg);
		reciveMessageQueue.offer(serviceSession);
	}
	
	public static void sendMessage(ChannelHandlerContext ctx, String msg) {
		if(msg == null || msg.length() == 0) return;
		//只是负责向消息队列中丢消息
		BaseServiceSession serviceSession = new BaseServiceSession(ctx, msg);
		sendMessageQueue.offer(serviceSession);
	}
}
